package com.xdja.kafka.hdfs.sink.writer.manage;

import com.xdja.kafka.hdfs.sink.writer.definition.WriterDefinition;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.Path;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.File;
import java.io.IOException;

/**
 * FSDataOutputStream  writer管理
 */
public class TextWriterManage extends AbstractWriterManage<FSDataOutputStream> {
    private static final Logger log = LoggerFactory.getLogger(TextWriterManage.class);

    public final static String txtSuffix = ".txt";
    /**
     * 实例
     */
    private static TextWriterManage instance;

    /**
     * 私有构造函数，为单例模式做准备
     */
    private TextWriterManage() {

    }

    /**
     * 单例模式，获取实例
     * @return
     */
    public static TextWriterManage getInstance() {
        if(instance == null) {
            instance = new TextWriterManage();
        }
        return instance;
    }

    @Override
    protected void clolseWriter(FSDataOutputStream closeFSDataOutputStream) {
        if(closeFSDataOutputStream != null) {
            try {
                closeFSDataOutputStream.hsync();
                closeFSDataOutputStream.close();
            } catch (IOException e) {
                log.error("关闭closeFSDataOutputStream失败，原始:{}", e.getMessage(), e);
            }
        }
    }

    /**
     * 文本格式的writer，存储路径  {hdfsPath}/{YYYY-MM}/{YYYY-MM-DD}/{YYYY-MM-DD}.txt
     * @param date
     * @return
     * @throws IOException
     */
    @Override
    protected FSDataOutputStream buildWriter(String date) throws IOException {
        String dateDirectory = getHdfsDateDir(date);
        Path dateDirectoryPath = new Path(dateDirectory);
        //日期目录不存在，创建
        if(!WriterDefinition.fileSystem.exists(dateDirectoryPath)) {
            WriterDefinition.fileSystem.mkdirs(dateDirectoryPath);
            log.info("创建了目录:{}", dateDirectory);
        }
        //日期文件
        Path dateFilePath = new Path(dateDirectory + File.separator + date + txtSuffix);
        //日期文件不存在，则创建
        if(!WriterDefinition.fileSystem.exists(dateFilePath)) {
            //若不存在则创建，支持并发模式：创建完要及时关闭，并在后面重新打开，否则有问题
            WriterDefinition.fileSystem.create(dateFilePath, false).close();
            log.info("hadoop创建了文件:{}", dateFilePath.getName());
        }
        //返回追加模式的writer
        FSDataOutputStream fsDataOutputStream = WriterDefinition.fileSystem.append(dateFilePath);
        log.info("成功创建ParquetWriter, date:{}", date);
        return fsDataOutputStream;
    }
}
